package edu.kit.ksri.db.mturk.simulations.correctness;

import java.io.BufferedReader;
import java.io.File;
import java.io.FileReader;
import java.io.FilenameFilter;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Calendar;
import java.util.Collections;
import java.util.LinkedList;
import java.util.Queue;
import java.util.Random;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;

import org.apache.log4j.Logger;

import com.amazonaws.mturk.requester.Assignment;

import edu.kit.ksri.db.mturk.plugins.display.BrowserDisplay;
import edu.kit.ksri.db.mturk.var.CorrSimMode;
import edu.kit.ksri.db.mturk.var.Icd;
import edu.kit.ksri.db.mturk.var.StaticInformation;

/**
 * Copyright (c) 2010-2011 Karlsruhe Institute of Technology (KIT), Germany
 *
 * Permission is hereby granted, free of charge, to any person obtaining a
 * copy of this software and associated documentation files (the "Software"),
 * to deal in the Software without restriction, including without limitation
 * the rights to use, copy, modify, merge, publish, distribute, sublicense,
 * and/or sell copies of the Software, and to permit persons to whom the
 * Software is furnished to do so, subject to the following conditions:
 * 
 * The above copyright notice and this permission notice shall be included in
 * all copies or substantial portions of the Software.
 * 
 * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
 * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
 * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
 * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
 * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
 * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER
 * DEALINGS IN THE SOFTWARE.
 * 
 * --------------------------------------------------------------------------
 * 
 * Software: CSP/WMV tool for dynamic quality management of microtasks
 * http://www.ksri.kit.edu/Upload/Users/PeopleClouds/HCOMP2011_An_extendable_toolkit_for_managing_quality_of_human-based_eServices.pdf
 * 
 * Organization: Karlsruhe Institute of Technology (KIT), Germany
 * http://www.kit.edu
 * 
 * Project: PeopleClouds project of the Karlsruhe Service Research Institute(KSRI)
 * http://www.ksri.kit.edu/Default.aspx?PageId=818
 * 
 * Contributing authors: David Bermbach, Sandra Rath, Pascal Wichmann, Robert Kern
 */
public class CorrSimMTurk implements Runnable{

	private static final Logger basiclog = Logger.getLogger("basic." + CorrSimMTurk.class);
	
	/** all hit ids */
	private static String[] hitIds;

	/** all results which can be retrieved */
	private static final ConcurrentHashMap<String, Queue<Assignment>> outputAnswers = new ConcurrentHashMap<String, Queue<Assignment>>();

	/**
	 * all queued hit ids
	 */
	private static final Queue<String> queuedHitIds = new ConcurrentLinkedQueue<String>();

	/** contains all qualification values for workers */
	private static final ConcurrentHashMap<String, Integer> qualificationValues = new ConcurrentHashMap<String, Integer>();
	
	/** contains all assignments for each hit */
	private static ConcurrentHashMap<String, LinkedList<Assignment>> workerAssignments = new ConcurrentHashMap<String, LinkedList<Assignment>>();

	/** determines the end of the simulation */
	private static boolean running = true;

	/** counts the number of added answers */
	private static int counter = 0;

	//hack for old XML files that have no assignment ID
//	private static int assignmentcounter = 0;

	/**
	 * initializes and starts the {@link CorrSimMTurk}
	 * 
	 * @param inputDirectory directory in which the all.HITID.xml files can be found
	 * 
	 */
	public static void initialize(String inputDirectory) throws Exception {
		//Read all the files from the input directory
		File dir = new File(inputDirectory);
		if(dir.isDirectory()){
			File[] allXmlFiles = dir.listFiles(new FilenameFilter(){
				public boolean accept(File dir, String name) {
					//Files have to start with "all.", otherwise the files should not be included
					if (name.trim().startsWith("all."))
						return true;
					return false;
				}
            });
			
			//Initialize all the possible HIT IDs
			hitIds = new String[allXmlFiles.length];
			
			//Do for every file
			for(File f : allXmlFiles){
				String hitid = "";
				try {
					//Read all lines in the XML file
					BufferedReader br = new BufferedReader(new FileReader(f));
					String content = "", line = br.readLine();
					while (line != null) {
						content += line;
						line = br.readLine();
					}
					
					//Read HIT ID
					int start = content.indexOf("hitid=") + 7;
					content = content.substring(start);
					int end = content.indexOf("\"");
					hitid = content.substring(0, end);
					
					//Read Mapping number
					start = content.indexOf("mappingNumber=") + 15;
					content = content.substring(start);
					end = content.indexOf("\"");
					int mappingNumber = Integer.parseInt(content.substring(0, end));
					
					//Add the HIT to the list of HITIDs for correct order
					hitIds[mappingNumber-1] = hitid; 
					
					//Read number of results
					start = content.indexOf("numberOfResults=") + 17;
					content = content.substring(start);
					end = content.indexOf("\"");
					int number = Integer.parseInt(content.substring(0, end));
										
					//Do that for every assignment in the XML file
					LinkedList<Assignment> assignments = new LinkedList<Assignment>();
					for (int i = 0; i < number; i++) {
						// Read worker ID
						start = content.indexOf("worker=") + 8;
						content = content.substring(start);
						end = content.indexOf("\"");
						String workerid = content.substring(0, end);
												
						//Read assignment ID
						start = content.indexOf("assignmentID=") + 14; 
						content= content.substring(start);
						end = content.indexOf("\"");
						String assignmentID = content.substring(0, end);
						
						//Read submission time
						start = content.indexOf("submissionTime=") + 16; 
						content= content.substring(start);
						end = content.indexOf("\"");
						content.substring(0, end);//submission time
					
						//hack for old XML files that have no assignment ID
//						assignmentID = ""+assignmentcounter++;
						
						//Need to concatenate the (different than originally read) XML answer for the assignment that has to be constructed
						String xml = null;
						
						if(StaticInformation.scenario instanceof Icd){
									
							//Concatenate the log entry
							String answersLog = "Reading siminput directory - Current HIT is " + hitid + " and assignment is " + assignmentID + ". Answers are ";
							String answers = "";
							
							//Concatenate the XML answer
							xml = "<?xml version=\"1.0\" encoding=\"UTF-8\"?><QuestionFormAnswers xmlns=\"http://mechanicalturk.amazonaws.com/AWSMechanicalTurkDataSchemas/2005-10-01/QuestionFormAnswers.xsd\">";
							
							//Do that for every single answer in the assignment (Consists of several labels * (several flags + 1 code) + 1 comment field
							for(int j=0; j < Icd.maxNumberOfLabels * (Icd.flagCount + 1) + 1; j++){
								//Read question ID
								start = content.indexOf("<QuestionIdentifier>") + 20;
								content = content.substring(start);
								end = content.indexOf("</QuestionIdentifier>");
								String questionId = content.substring(0, end);
								basiclog.debug(questionId);

								//Read the answer
								start = content.indexOf("<FreeText>") + 10;
								content = content.substring(start);
								end = content.indexOf("</FreeText>");
								String answer = content.substring(0, end);
								basiclog.debug(answer);
								
								//Concatenate the answers for log
								if(j == (Icd.maxNumberOfLabels * Icd.flagCount)-1)
									answers = answers + answer;
								else
									answers = answers + answer + ", ";

								content = content.substring(end);
								
								//Concatenate the XML
								xml = xml + "<Answer><QuestionIdentifier>"+questionId+"</QuestionIdentifier><FreeText>"+answer+"</FreeText></Answer>";
							}
							
							//Concatenate the XML
							xml = xml + "</QuestionFormAnswers>";
							
							//Log what we just read
							basiclog.info(answersLog + answers);
						}
						else{
							//OCR and Poker consist only of one Question Identifier and one FreeText
							//Read question ID
							start = content.indexOf("<QuestionIdentifier>") + 20;
							content = content.substring(start);
							end = content.indexOf("</QuestionIdentifier>");
							String questionId = content.substring(0, end);
	
							//Read the answer
							start = content.indexOf("<FreeText>") + 10;
							content = content.substring(start);
							end = content.indexOf("</FreeText>");
							String answer = content.substring(0, end);
	
							content = content.substring(end);
							
							//Log what we just read
							basiclog.info("Reading siminput directory - Current HIT is " + hitid + " and assignment is " + assignmentID + ". Answer is " + answer);
							
							//Concatenate the XML answer
							xml = "<?xml version=\"1.0\" encoding=\"UTF-8\"?><QuestionFormAnswers xmlns=\"http://mechanicalturk.amazonaws.com/AWSMechanicalTurkDataSchemas/2005-10-01/QuestionFormAnswers.xsd\"><Answer>"+
							"<QuestionIdentifier>"+questionId+"</QuestionIdentifier><FreeText>"+answer+"</FreeText></Answer></QuestionFormAnswers>";
						}
						
						//Add worker with qualification value
						//1 means the worker may work because of right qualification
						qualificationValues.put(workerid, new Integer(1));
						
						//Create new assignment
						Assignment assignment = new Assignment();
						assignment.setAnswer(xml);
						assignment.setWorkerId(workerid);
						assignment.setAssignmentId(assignmentID);
						assignment.setHITId(hitid);
						
						//Add the assignment to the assignment collection
						assignments.add(assignment);
					}
					//Add the assignment collection to the possible worker assignment collection
					workerAssignments.put(hitid, assignments);

				} catch (Exception e) {
					basiclog.error("Error while reading file " + f);
				}
			}

			//We are done with the initializing, now start the actual simulation
			running = true;
			Thread t = new Thread(new CorrSimMTurk());
			t.start();
			
		}
		else{
			basiclog.error("SIMINPUT directory is not a directory.");
		}
	}
	
	public void run(){
		while(running){
			;			
			//Get the next HIT we should process
			String hit = null;
			try {
				hit = getNextHIT();
			} catch (Exception e) {
				try {
					//No new HIT there, so sleep a bit and try again
					Thread.sleep(500);
					continue;
				} catch (InterruptedException e1) {
					basiclog.error("InterruptedException occurred: " + e1);
				}
			}

			//Stores all the assignments for a single HIT
			LinkedList<Assignment> assignmentsForHit = workerAssignments.get(hit);

			//Do this until we break free
			while(true){
				//How many assignments do we have at most?
				int maxAssignmentCount = assignmentsForHit.size();
				if(maxAssignmentCount == 0){
					//Too bad, we do not have enough assignments
					basiclog.error("Not enough assignments for HIT " + hit + " existing.");
					BrowserDisplay.update(hit);
					//Continue with the next HIT
					break;
				}
				
				//Pick an assignment
				Assignment assignment = null;
				int random = 0;
				
				if(StaticInformation.scenario.getCorrSimModeAssignment() == CorrSimMode.RANDOM){
					//Pick a random assignment
					random = new Random().nextInt(maxAssignmentCount);
					assignment = assignmentsForHit.get(random);
					basiclog.debug("Random/MaxAssignmentCount: " + (random+1) + "/" + maxAssignmentCount);
				}
				else if(StaticInformation.scenario.getCorrSimModeAssignment() == CorrSimMode.ORDER){
					//Pick the assignment in order
					assignment = assignmentsForHit.getFirst();
				}
				else{
					basiclog.error("Pick a valid CorrSimMode.");
				}
				
				String workerId = assignment.getWorkerId();
				
				//Check if worker has the correct qualification to work
				if(!qualificationValues.isEmpty()){
					//Get the latest qualification
					int newQuali = CorrSimMTurk.getQualificationUpdate(workerId);
					
					if (newQuali >= StaticInformation.activePoolQualificationValue){
						//Yay, worker is allowed to work
						if(StaticInformation.scenario.getCorrSimModeAssignment() == CorrSimMode.RANDOM){
							assignmentsForHit.remove(random);
						}
						else if(StaticInformation.scenario.getCorrSimModeAssignment() == CorrSimMode.ORDER){
							assignmentsForHit.removeFirst();
						}
						else{
							basiclog.error("Pick a valid CorrSimMode.");
						}
						//Worker is working - we can add their answer
						if (maxAssignmentCount == 10) {
							basiclog.error("New HIT - wait for some time.");
							try {
								Thread.sleep(500);
							} catch (InterruptedException e) {
								e.printStackTrace();
							}
						}
							
						addAnswer(assignment.getHITId(), assignment);
						
						//We found an assignment for the HIT, so we can continue with the next HIT
						break;
					}
					else{
						//Look for another worker with a qualification that is good enough
						basiclog.info("Worker " + workerId + " has not the right qualification.");
					}
				}
			}
		}
		
	}

	/**
	 * @return the running
	 */
	public static boolean isRunning() {
		return running;
	}

	/**
	 * @param e
	 * @return
	 * @see java.util.Queue#add(java.lang.Object)
	 */
	public static boolean addHIT(String hitId) {
		basiclog.info("Adding hit number " + hitId);
		outputAnswers.put(hitId, new ConcurrentLinkedQueue<Assignment>());
		return queuedHitIds.add(hitId);
	}

	public static boolean extendHIT(String hitId) {
		try {
			synchronized (queuedHitIds) {
				ArrayList<String> temp = new ArrayList<String>();
				while (queuedHitIds.size() > 0) {
					temp.add(queuedHitIds.remove());
				}
				queuedHitIds.add(hitId);
				for (String ee : temp) {
					queuedHitIds.add(ee);
				}
				if(!outputAnswers.containsKey(hitId)){
					outputAnswers.put(hitId, new ConcurrentLinkedQueue<Assignment>());
				}
				basiclog.info("HIT "+hitId+" extended.");
				return true;
			}
		} catch (Exception e1) {
			e1.printStackTrace();
			basiclog.warn("HIT "+hitId+" not extended.");
			return false;
		}
	}

	/**
	 * @return
	 * @see java.util.Collection#isEmpty()
	 */
	public static boolean hasHITs() {
		return queuedHitIds.isEmpty();
	}

	/**
	 * @return
	 * @see java.util.Queue#remove()
	 */
	public static String getNextHIT() {
		String temp = queuedHitIds.remove();
		basiclog.info("Retrieving hit number " + temp + " (further "
				+ queuedHitIds.size() + " HITs are queued right now.)");
		return temp;
	}

	/**
	 * @param e
	 * @return
	 * @see java.util.Queue#add(java.lang.Object)
	 */
	public static boolean addAnswer(String hitId, Assignment assignment) {
		counter++;
		assignment.setSubmitTime(Calendar.getInstance());
		basiclog.info("Adding answer for HIT " + hitId + " (" + counter
				+ " answers in total).");
		/*try {
			//Thread.sleep(4000);
			Thread.sleep(500);
			} catch (InterruptedException e) {
			e.printStackTrace();
		} */
		return outputAnswers.get(hitId).add(assignment);
	}

	/**
	 * @return
	 * @see java.util.Collection#isEmpty()
	 */
	public static boolean hasAnswers() {
		return outputAnswers.isEmpty();
	}

	/**
	 * @return
	 * @see java.util.Collection#remove()
	 */
	public static Assignment getNextAnswerForHit(String hitId) {
		return outputAnswers.get(hitId).remove();
	}

	/**
	 * @param key
	 * @param value
	 * @return
	 * @see java.util.Map#put(java.lang.Object, java.lang.Object)
	 */
	public static Integer addQualificationUpdate(String key, Integer value) {
		if(qualificationValues.containsKey(key)){
			return qualificationValues.replace(key, value);
		}
		else{
			return qualificationValues.put(key, value);
		}
	}

	/**
	 * @param key
	 * @return
	 * @see java.util.Map#remove(java.lang.Object)
	 */
	public static Integer getQualificationUpdate(Object key) {
			return qualificationValues.get(key);
	}

	/**
	 * Returns all possible HIT IDs
	 * @return hitIds
	 */
	public static String[] getHitIds() {
		if(StaticInformation.scenario.getCorrSimModeHIT() == CorrSimMode.RANDOM){
			Collections.shuffle(Arrays.asList(hitIds));
		}
		return hitIds;
	}
	
	
}
